package cn.sheep.dmp.utils

import org.apache.commons.lang3.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.Row

import scala.collection.mutable

/**
  * sheep.Old @ 64341393
  * Created 2018/4/1
  */
object TUtils {


    val hasUserIds: String =
        """
          |isNotEmpty(imei) OR isNotEmpty(idfa) OR isNotEmpty(mac) OR isNotEmpty(androidid) OR isNotEmpty(openudid) OR
          |isNotEmpty(imeimd5) OR isNotEmpty(idfamd5) OR isNotEmpty(macmd5) OR isNotEmpty(androididmd5) OR isNotEmpty(openudidmd5) OR
          |isNotEmpty(imeisha1) OR isNotEmpty(idfasha1) OR isNotEmpty(macsha1) OR isNotEmpty(androididsha1) OR isNotEmpty(openudidsha1)
        """.stripMargin


    /**
      * 获取用户某一个不为空的id
      */
    def getUserId(row: Row) = {

        row match {
            case v if StringUtils.isNotEmpty(v.getAs[String]("imei")) => "IM#"+v.getAs[String]("imei")
            case v if StringUtils.isNotEmpty(v.getAs[String]("idfa")) => "ID#"+v.getAs[String]("idfa")
            case v if StringUtils.isNotEmpty(v.getAs[String]("mac")) => "MC#"+v.getAs[String]("mac")
            case v if StringUtils.isNotEmpty(v.getAs[String]("androidid")) => "AD#"+v.getAs[String]("androidid")
            case v if StringUtils.isNotEmpty(v.getAs[String]("openudid")) => "OP#"+v.getAs[String]("openudid")

            case v if StringUtils.isNotEmpty(v.getAs[String]("imeimd5")) => "IMM#"+v.getAs[String]("imeimd5")
            case v if StringUtils.isNotEmpty(v.getAs[String]("idfamd5")) => "IDM#"+v.getAs[String]("idfamd5")
            case v if StringUtils.isNotEmpty(v.getAs[String]("macmd5")) => "MCM#"+v.getAs[String]("macmd5")
            case v if StringUtils.isNotEmpty(v.getAs[String]("androididmd5")) => "ADM#"+v.getAs[String]("androididmd5")
            case v if StringUtils.isNotEmpty(v.getAs[String]("openudidmd5")) => "OPM#"+v.getAs[String]("openudidmd5")

            case v if StringUtils.isNotEmpty(v.getAs[String]("imeisha1")) => "IMS#"+v.getAs[String]("imeisha1")
            case v if StringUtils.isNotEmpty(v.getAs[String]("idfasha1")) => "IDS#"+v.getAs[String]("idfasha1")
            case v if StringUtils.isNotEmpty(v.getAs[String]("macsha1")) => "MCS#"+v.getAs[String]("macsha1")
            case v if StringUtils.isNotEmpty(v.getAs[String]("androididsha1")) => "ADS#"+v.getAs[String]("androididsha1")
            case v if StringUtils.isNotEmpty(v.getAs[String]("openudidsha1")) => "OPS#"+v.getAs[String]("openudidsha1")

        }

    }

    /**
      * 获取当前行的用户便签数据，返回当前行的一组标签数据
      * @param row
      * @return
      */
    def getCurrentLineTags(row: Row,
                           appdict: Broadcast[Map[String, String]],
                           stopdict: Broadcast[Map[String, Null]]) : mutable.HashMap[String, Int]= {

        val map = new mutable.HashMap[String, Int]()

        val adSpt = row.getAs[Int]("adspacetype")
        if (adSpt>0 && adSpt < 10) map.put("LC0" + adSpt, 1) else map.put("LC" + adSpt, 1)
        val adSptn = row.getAs[String]("adspacetypename")
        if (StringUtils.isNotEmpty(adSptn)) map.put("LN" + adSptn, 1)


        val appId = row.getAs[String]("appid")
        val appName = row.getAs[String]("appname")

        if (StringUtils.isEmpty(appName)) {
            if (StringUtils.isNotEmpty(appId)) {
                // 从广播变量中获取
                val newAppName = appdict.value.getOrElse(appId, appId)
                map.put("APP"+newAppName, 1)
            }
        } else map.put("APP"+appName, 1)

        // 渠道
        val channelId = row.getAs[Int]("adplatformproviderid")
        if (channelId > 0) map.put("CN"+channelId, 1)

        // 设备
        val client = row.getAs[Int]("client")
        //获取设备联网方式字段
        val net = row.getAs[String]("networkmannername").toUpperCase
        //获取运营商名称字段
        val opera = row.getAs[String]("ispname")

        client match {
            case 1 => map.put("D00010001", 1)
            case 2 => map.put("D00010002", 1)
            case 3 => map.put("D00010003", 1)
            case _ => map.put("D00010004", 1)
        }

        net match {
            case "WIFI" => map.put("D00020001", 1)
            case "4G" => map.put("D00020002", 1)
            case "3G" => map.put("D00020003", 1)
            case "2G" => map.put("D00020004", 1)
            case _ => map.put("D00020005", 1)
        }

        opera match {
            case "移动" => map.put("D00030001", 1)
            case "联通" => map.put("D00030002", 1)
            case "电信" => map.put("D00030003", 1)
            case _ => map.put("D00030004", 1)
        }


        //获取关键字字段
        val keywords = row.getAs[String]("keywords")
        if (StringUtils.isNotEmpty(keywords)) {
            keywords.split("\\|")
              .filter(word => word.length >= 3 && word.length <= 8) // 长度的符合要求
              .filter(word => !stopdict.value.contains(word)) // 判断是否是否在停用词库中，如果存在则不要
              .foreach(word => map.put("K"+word, 1))
        }


        // 地域标签
        val pName = row.getAs[String]("provincename")
        val cName = row.getAs[String]("cityname")

        if (StringUtils.isNotEmpty(pName)) map.put("ZP"+pName, 1)
        if (StringUtils.isNotEmpty(cName)) map.put("ZC"+cName, 1)

        map
    }

}
